package com.wsg.dep.queue;

import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.TimeoutException;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import com.wsg.dep.core.event.EventFactory;
import com.wsg.dep.demo.LongEvent;
import com.wsg.dep.demo.LongEventFactory;
import com.wsg.dep.demo.LongEventHandler;

import java.text.DecimalFormat;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @className: Test
 * @author: 承泽
 * @date: 2025/3/17 20:40
 * @description:
 */
public class Test {
    public static final int total = 1000 * 50000;
    /**
     * 8：30.23百万/s
     * 16：50.607百万/s
     * 32：80.108百万/s
     * 64：108.46百万/s
     * 128：68.587百万/s
     *
     * 最理想的状态是参数为64或32的时候，不要迷信的相信设置的越大吞吐量就越高。
     *
     */
    public static final int bufferSize = 64;

    public static void main(String[] args) {

        new Thread(() -> {
            countNN();
        }, "countNN-单线程内遍历处理").start();

        new Thread(() -> {
            countS();
        }, "countS-创建队列后单线程内获取队列数据处理").start();
        new Thread(() -> {
            countPs();
        }, "countPs-自实现的高性能队列方式").start();
        new Thread(() -> {
            countD();
        }, "countD-Disruptor的默认实现").start();


    }


    public static void countPs() {
        // 初始化队列和门控
        BaseRingQueue<LongEvent> ringBuffer = new BaseRingQueue<>(
                new ProducerConsumerModel.LongEventFactory(), bufferSize
        );
        ProducerConsumerBarrier barrier = new ProducerConsumerBarrier(bufferSize);

        // 自定义线程工厂，为线程设置名字
        ThreadFactory namedThreadFactory = new ThreadFactory() {
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("countPs-自实现的高性能队列方式-" + threadNumber.get());
                return thread;
            }
        };
        // 创建线程池，并使用自定义线程工厂
        ExecutorService executor = new ThreadPoolExecutor(
                1 + 1,
                1 + 1,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(),
                namedThreadFactory
        );

        // 启动生产者
        for (int i = 0; i < 1; i++) {
            executor.submit(new ProducerConsumerModel.Producer(ringBuffer, barrier, i));
        }

        // 启动消费者
        for (int i = 0; i < 1; i++) {
            executor.submit(new ProducerConsumerModel.Consumer(ringBuffer, barrier, i));
        }
    }

    public static void doSomeThing(LongEvent longEvent) {
        if (longEvent != null && longEvent.get() >= total - 1) {
            long time = System.currentTimeMillis() - longEvent.getTime();
            double sendsTime = time / 1000.0;
            if (sendsTime < 1e-3) sendsTime = 1.0; // 防止除零

            // 计算吞吐量
            double avg = total / sendsTime;

            // 数字格式化器配置
            DecimalFormat[] formatters = {
                    new DecimalFormat("#,###"),           // 格式1: 千分位 (1,234,567)
                    new DecimalFormat("#.###E0"),         // 格式2: 科学计数法 (1.235E6)
                    new DecimalFormat("#.###'千/s'"),       // 格式3: 千级单位
                    new DecimalFormat("#.###'百万/s'")        // 格式4: 百万级单位
            };

            // 动态选择格式
            String avgText;
            if (avg < 1_000) {
                avgText = formatters[0].format(avg);  // 普通格式
            } else if (avg < 1_000_000) {
                avgText = formatters[2].format(avg / 1_000);  // 转换为千单位
            } else if (avg < 1_000_000_000) {
                avgText = formatters[3].format(avg / 1_000_000);  // 转换为百万单位
            } else {
                avgText = formatters[1].format(avg);  // 科学计数法
            }

            // 构造输出信息
            String output = String.format(
                    "\n[Benchmark Result]\n" +
                            "| Processed: %d\n" +
                            "| Thread: %s\n" +
                            "| Cost: %,d ms\n" +
                            "| Throughput: %s\n" +
                            "-------------------------",
                    longEvent.get(),
                    Thread.currentThread().getName(),
                    time,
                    avgText
            );

            System.out.println(output);
        }
    }


    public static void countNN() {
        LongEventFactory factory = new LongEventFactory();

        // 队列大小
        for (int j = 0; j < total; j++) {
            // 模拟数据处理
            LongEvent longEvent = factory.newInstance();
            longEvent.set(j);
            doSomeThing(longEvent);
        }
    }

    public static void countS() {
        EventFactory eventFactory = new EventFactory<LongEvent>() {
            private long time = System.currentTimeMillis();

            @Override
            public LongEvent newInstance() {
                LongEvent longEvent = new LongEvent();
                longEvent.setTime(time);
                return longEvent;
            }
        };
        BaseRingQueue<LongEvent> queue = new BaseRingQueue<>(eventFactory, bufferSize);
        for (int j = 0; j < total; j++) {
            LongEvent longEvent = queue.get(j);
            longEvent.set(j);

            // 模拟数据处理
            doSomeThing(longEvent);
        }
    }


    public static void countD() {
        // 1. 创建事件工厂
        LongEventFactory factory = new LongEventFactory();

        // 2. 指定RingBuffer大小（必须是2的幂次方）

        // 3. 创建Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(
                factory,
                bufferSize,
                DaemonThreadFactory.INSTANCE,
                ProducerType.MULTI, // 单生产者模式
                new BlockingWaitStrategy()
        );

        // 4. 注册事件处理器
        disruptor.handleEventsWith(new LongEventHandler());

        // 5. 启动Disruptor
        disruptor.start();

        // 6. 发布事件
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        for (long i = 0; i < total; i++) {
            long sequence = ringBuffer.next(); // 获取下一个可用序列号
            try {
                LongEvent event = ringBuffer.get(sequence); // 获取事件对象
                event.set(i); // 设置事件数据
            } finally {
                ringBuffer.publish(sequence); // 发布事件
            }
        }
//        try {
//            disruptor.shutdown(15, TimeUnit.SECONDS);
//        } catch (TimeoutException e) {
//            e.printStackTrace();
//        }
    }
}
